package com.catmiao.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable
import scala.util.Random

/**
 * @title: SparkStreaming01_WordCount
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/3/25 00:31
 */
object SparkStreaming03_DIY_Receiver {

  def main(args: Array[String]): Unit = {


    // 创建环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //  param1 : 环境配置，SparkConf
    //  param2 ： 采集周期【批量处理周期】
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    // 自定义采集器
    val inputDStream: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver)

    inputDStream.print()

    // 1. 启动采集器
    ssc.start()



    // 2. 等待采集器的关闭
    ssc.awaitTermination()
  }


  /**
   * 自定义采集器
   *  - 采集的数据类型
   *  - 数据存储方式
   *
   *
   */
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {

    private var flag = true;

    // 启动时执行的方法
    override def onStart(): Unit = {

        new Thread(
          new Runnable {
            override def run(): Unit = {
              while(flag){
                val msg = "采集的到的数据:" + new Random().nextInt(10).toString
                store(msg) // 存储数据
                Thread.sleep(500)
              }
            }
          }
      ).start()
    }


    // 停止时执行的方法
  override def onStop(): Unit = {
    flag = false
  }
}
}
